package com.fot.config;

import com.fot.annotation.RedisMQMessageListener;
import com.fot.core.ConsumeWorker;
import com.fot.core.listener.RedisMQListener;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RedissonClient;
import org.springframework.aop.framework.AopProxyUtils;
import org.springframework.core.env.StandardEnvironment;

import java.util.Arrays;

/**
 * @author 殷豪
 */
@Slf4j
public class ListenerContainerConfiguration {
    private StandardEnvironment environment;
    private RedisMQProperties redisMQProperties;
    private final ConsumeWorker consumeWorker;

    public ListenerContainerConfiguration(StandardEnvironment environment,
                                          RedisMQProperties redisMQProperties,
                                          RedissonClient redissonClient) {
        this.environment = environment;
        this.redisMQProperties = redisMQProperties;
        consumeWorker = new ConsumeWorker(ConsumeWorker.class.getName(), redissonClient, redisMQProperties);
    }

    public void registerContainer(String beanName, Object bean, RedisMQMessageListener annocation) {
        Class<?> clazz = AopProxyUtils.ultimateTargetClass(bean);
        if (!RedisMQListener.class.isAssignableFrom(bean.getClass())) {
            throw new IllegalStateException(clazz + "is not instance of " + RedisMQListener.class.getName());
        }
        String topic = this.environment.resolvePlaceholders(annocation.topic());
        boolean listenerEnabled = redisMQProperties.getConsumer().getListeners().getOrDefault(topic, true);
        if (!listenerEnabled) {
            log.debug("Consumer listener (name:{}, topic:{}) is not enabled by configuration, will ignore initialization.", beanName, topic);
            return;
        }
        consumeWorker.addListeners(topic, Arrays.asList((RedisMQListener<?>) bean));
    }
}
